package com.shujia.spark.mllib

import org.apache.spark.ml.feature.LabeledPoint
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.regression.{LinearRegression, LinearRegressionModel}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Demo2Line {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("line")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    import spark.implicits._

    /**
      * 读取原始的数据
      *
      */
    val linesDS: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("y DOUBLE, x DOUBLE")
      .load("data/line.txt")

    /**
      * 将数据转换成算法可以识别的向量
      *
      */
    val labelPointDF: Dataset[LabeledPoint] = linesDS.map {
      case Row(y: Double, x: Double) =>
        //将数据转换成标记点，y作为目标值，x作为特征向量
        LabeledPoint(y, Vectors.dense(x))
    }

    val labelPointDAta: DataFrame = labelPointDF.toDF()
    labelPointDAta.printSchema()
    labelPointDAta.show()

    /**
      * 构建算法，将数据带入算法训练模型
      *
      */


    //线性回归算法
    val lr: LinearRegression = new LinearRegression()

    /**
      * 将数据带入算法训练模型
      * 模型：特征的权重和截距
      *
      */
    val model: LinearRegressionModel = lr.fit(labelPointDAta)


    //截距
    val intercept: Double = model.intercept
    println(intercept)
    //权重
    println(model.coefficients)

    //将新的x带入模型计算得到未知的y
    val y: Double = model.predict(Vectors.dense(7.1))
    println(y)


  }

}
